package com.example.toy.core.commit;

import com.example.toy.core.version.ToyVersion;
import lombok.SneakyThrows;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.ByteArrayInputStream;
import java.util.Optional;

/**
 * FsOfflineToyCommitter
 *
 * @author shenbing
 * @date 2021/12/23
 */
public class FsOfflineToyCommitter extends FsOfflineRealtimeToyCommitView
    implements ToyCommitter, ToyCommitView {

  protected final FileSystem fs;

  @SneakyThrows
  public FsOfflineToyCommitter(Path basePath) {
    super(basePath);
    this.fs = basePath.getFileSystem(new Configuration());

    if (!fs.exists(this.commitDirPath)) {
      fs.mkdirs(this.commitDirPath);
    }
  }

  @Override
  public void submit(CommitMsg commit) {
    submit(commit, commit.getCommitState());
  }

  @SneakyThrows
  @Override
  public void submit(CommitMsg commit, CommitState commitState) {
    Optional<CommitMsg> exists = getCommit(commit.getVersion());
    if (exists.isPresent()) {
      throw new IllegalStateException("Commit version already exists: " + commit.getVersion());
    }
    Path commitFileName = getCommitFileName(commit.getVersion(), commitState);
    byte[] bytes = commit.serialize();
    try (FSDataOutputStream fsDataOutputStream = fs.create(commitFileName, true);
        ByteArrayInputStream in = new ByteArrayInputStream(bytes)) {
      IOUtils.copy(in, fsDataOutputStream);
    }
  }

  private Path getCommitFileName(ToyVersion version, CommitState commitState) {
    return new Path(commitDirPath, version.getTimestamp() + "." + commitState);
  }

  @SneakyThrows
  @Override
  public void changeState(ToyVersion version, CommitState state) {
    Optional<CommitMsg> commit = getCommit(version);
    if (commit.isPresent()) {
      CommitState commitState = commit.get().getCommitState();
      if (commitState.supportTransform(state)) {
        Path commitFileName = getCommitFileName(commit.get().getVersion(), commitState);
        fs.rename(commitFileName, getCommitFileName(commit.get().getVersion(), state));
      } else {
        throw new IllegalStateException(
            "Can not transform version from state " + commitState + " to " + state);
      }

    } else {
      throw new IllegalStateException("version not exists: " + version);
    }
  }

  @Override
  public void compact() {}
}
